package kafkaStudy2; 

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/** 
* @author 朱宏亮
* @version 创建时间：2017年12月13日 下午2:15:26 
* 类说明 直接使用 kafka_2.11-1.0.0，未整合spring
*/
public class ConsumerGroup {

    @SuppressWarnings("resource")
	public static void main(String[] args) {
         
         String topic = "zhlTopic";//args[0].toString();
         String group = "GroupZHL01";//args[1].toString();
         Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("client.id", "ClientZHL001");
         props.put("group.id", group);
         props.put("enable.auto.commit", "true");
         props.put("auto.commit.interval.ms", "1000");
         props.put("session.timeout.ms", "30000");
         props.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         KafkaConsumer<Integer, String> consumer = new KafkaConsumer<Integer, String>(props);
         
         consumer.subscribe(Arrays.asList(topic));
         System.out.println("Subscribed to topic " + topic);
            
         while (true) {
            ConsumerRecords<Integer, String> records = consumer.poll(100);
               for (ConsumerRecord<Integer, String> record : records){
            	   System.out.println("=============ConsumerTest开始消费=============");
	               String topic1 = record.topic();
	               Integer key = record.key();
	               String value = record.value();
	               long offset = record.offset();
	               int partition = record.partition();
	               System.out.println("-------------topic:"+topic1);
	               System.out.println("-------------value:"+value);
	               System.out.println("-------------key:"+key);
	               System.out.println("-------------offset:"+offset);
	               System.out.println("-------------partition:"+partition);
	               System.out.println("~~~~~~~~~~~~~ConsumerTest消费结束~~~~~~~~~~~~~");
               }
            	  
         }     
      }  
    


}
 